package cn.aposoft.mq.kafka.hello;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class AutoCommitConsumerDemo {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "aposoft.cn:10091");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "false");
        props.put("auto.commit.interval.ms", "1000");
        props.put("max.poll.records", 10);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        @SuppressWarnings("resource")
        KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("quickstart-events"));
        for (int i = 0; i < 100; /* i++ */) {

            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100L));

            for (ConsumerRecord<String, byte[]> record : records) {
                i++;
                System.out.printf("value = %s, offset = %d, key = %s, %n", //
                        new String(record.value()), record.offset(), record.key());

                Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                TopicPartition partition = new TopicPartition(record.topic(), record.partition());
                OffsetAndMetadata offset = new OffsetAndMetadata(record.offset() + 1);// next fetch offset = current
                // offset + 1
                offsets.put(partition, offset);
                //consumer.commitSync(offsets);
                consumer.commitAsync(offsets, (v, e) -> System.out.println(v + "is commited async"));
            }

        }
//        consumer.unsubscribe();
        consumer.close();
    }
}
